package com.bigdata.mapreduce.topn;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Partitioner;

public class TPartitioner extends Partitioner<Tkey, IntWritable> {

    /**
     * 重写分区器，按照年月进行分区，相同年月的数据到一个reduceTask中
     *
     * @param tkey          待分区的key
     * @param intWritable   待分区的value
     * @param numPartitions reduceTask的数量，也就是分区数
     * @return
     */
    @Override
    public int getPartition(Tkey tkey, IntWritable intWritable, int numPartitions) {
        // 让key更加的散列
        int day = tkey.getDay();
        String mark = "";
        if (day <= 10) {
            mark = "_1";
        } else if (day <= 20) {
            mark = "_2";
        } else if (day <= 31) {
            mark = "_3";
        }
        // 2019-6_2, 是不是把相同年、月的数据就打散了，分布到不同的reduceTask
        // 按照相同的年、月分组，最终的结果是， 6-11~6-20好的温度的top2，每个月经过MR只会写出6条数据
        // 对上面的计算结果，再次做MR计算，输入的每个月的数据只有6条
        // 第二个MR程序要按照正确的分区逻辑来分区，按照年、月分区（key:2020-06_1）, 每个月只有6条数据，就不会有数据倾斜了
        // 一个MR解决不掉，那么就用2个，第一个MR的作用是在没有数据倾斜的基础上通过reduce聚合减少数据量
        // 通过2个没有数据倾斜的MR程序，从而解决数据倾斜的问题
//        return ("" + tkey.getYear() + tkey.getMonth() + mark).hashCode() % numPartitions;
        return ("" + tkey.getYear() + tkey.getMonth()).hashCode() % numPartitions;
    }
}
